/**
 * Copyright 2020 Inc.
 **/
package com.myz.cloud.rocketmq.producer.mq;

import com.myz.cloud.rocketmq.config.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author maoyz0621 on 2020/8/5
 * @version v1.0
 */
@Component
@Slf4j
public class MqProducer {

    /**
     * springboot提供
     */
    @Autowired
    RocketMQTemplate rocketMQTemplate;

    /**
     * rocketmq-client提供
     */
    @Autowired
    DefaultMQProducer defaultMQProducer;

    /**
     * send message synchronously
     */
    public void sendMsg(String topic, String msg) {
        rocketMQTemplate.convertAndSend(topic, msg);
    }

    /**
     * send spring message
     */
    public void send(String topic, String msg) {
        rocketMQTemplate.send(topic, MessageBuilder.withPayload(msg).build());
    }

    /**
     * 获取发送状态
     *
     * @param topic
     * @param order
     */
    public void syncSend(String topic, Order order) {
        SendResult send = rocketMQTemplate.syncSend(topic, order);
        log.info("{},send={}", send, send.getSendStatus());
    }

    /**
     * 发送一次
     */
    public void sendOneWay(String topic, Order order) {
        rocketMQTemplate.sendOneWay(topic, order);
    }

    /**
     * 异步发送send messgae asynchronously
     */
    public void asyncSend(String topic, Order order) {
        rocketMQTemplate.asyncSend(topic, order, new SendCallback() {

            /**
             * 接收发送成功到MQ的消息
             * @param sendResult
             */
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送成功:{}", sendResult);
            }

            /**
             * 发送失败
             * @param throwable
             */
            @Override
            public void onException(Throwable throwable) {
                log.error("发送失败", throwable);
            }
        });

    }

    /**
     * 事务消息
     * 事务消息不支持延时消息和批量消息。
     */
    public void sendTransaction(String txProducerGroup, String topic, Object order, String rocketHeader, String rocketHeaderVal) {
        // In sendMessageInTransaction(), the first parameter transaction name ("test")
        // must be same with the @RocketMQTransactionListener's member field 'transName'
        TransactionSendResult transaction = rocketMQTemplate.sendMessageInTransaction(txProducerGroup,
                topic,
                MessageBuilder.withPayload(order).setHeader(rocketHeader, rocketHeaderVal).build(),
                "aaaa");
        LocalTransactionState localTransactionState = transaction.getLocalTransactionState();
        log.info("事务消息:msgId = {},send = {}, status = {}\r\n", transaction.getMsgId(), localTransactionState, transaction.getSendStatus());
    }

    /**
     * 发送顺序消息 Send messages orderly
     */
    public void syncSendOrderly(Object msg, String hashKey) {
        // 设置队列选择器,hashkey，同一组放置同一个队列中消费
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                System.out.println("hashkey = " + o);
                return list.get(Integer.parseInt(o.toString()));
            }
        });
        rocketMQTemplate.syncSendOrderly("orderly_topic", MessageBuilder.withPayload(msg).build(), hashKey);
    }

    public void sendMsgClient(String topic, String msg) {
        Message message = new Message();
        try {
            SendResult send = defaultMQProducer.send(message);
            log.info("send={}", send.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}